package com.eshipenet.shipnet.equipmentdatahub

import java.util.Properties

import org.apache.flink.api.java.tuple.Tuple2
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.joda.time.Period
import play.api.libs.json.{JsObject, JsValue, Json}

import scala.util.Try

object EquipmentStreamProvider {

  def createEquipmentStream(env: StreamExecutionEnvironment): DataStream[JsValue] = {
    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "kafka-server:9092")
    // only required for Kafka 0.8
    //    properties.setProperty("zookeeper.connect", "localhost:2181")
    properties.setProperty("group.id", "equipmentdatahub.EquipmentWithModel")
    val der =
      new RawKafkaDeserializer()
    val consumer = new FlinkKafkaConsumer[RawKafkaMessage]("EquipmentWithModel", der, properties)
    consumer.setStartFromEarliest()
    val stream = env
      .addSource(consumer)
      .map { msg =>
        Try(Json.parse(msg.value))
      }
      .filter(_.isSuccess)
      .map(_.get)
    //      .map(m => (m.id, m))
    stream
  }
}
